package com.ehualu.impda.kafkasecurity;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;


public class security {
    private static final Logger LOG = LoggerFactory.getLogger(security.class);

    // 用户自己申请的机机账号keytab文件名称
    private static final String USER_KEYTAB_FILE = "user.keytab";
    // 用户自己申请的机机账号名称
    private static final String USER_PRINCIPAL = "super_rj";

    /**
     * 判断文件是否存在
     */
    private static boolean isFileExists(String fileName) {
        File file = new File(fileName);

        return file.exists();
    }


    /**
     * 判断是否为安全模式
     */
    private static Boolean isSecurityModel() {
        Boolean isSecurity = false;
        String krbFilePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator + "kafkaSecurityMode";

        Properties securityProps = new Properties();

        if (!isFileExists(krbFilePath)) {
            return isSecurity;
        }

        try {
            securityProps.load(new FileInputStream(krbFilePath));
            if ("yes".equalsIgnoreCase(securityProps.getProperty("kafka.client.security.mode"))) {
                isSecurity = true;
            }
        } catch (Exception e) {
            LOG.info("The Exception occured : {}.", e);
        }

        return isSecurity;
    }

    public static void securityPrepare() throws IOException {
//        String filePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator;
        String filePath = System.getProperty("user.dir") + File.separator + "resources" + File.separator;
        String krbFile = filePath + "krb5.conf";
        System.out.println("krbFile : " + krbFile);
        String userKeyTableFile = filePath + USER_KEYTAB_FILE;
        System.out.println("userKeyTableFile : " + userKeyTableFile);

        //windows路径下分隔符替换
        userKeyTableFile = userKeyTableFile.replace("\\", "\\\\");
        krbFile = krbFile.replace("\\", "\\\\");

        LoginUtilKafka.setKrb5Config(krbFile);
        LoginUtilKafka.setZookeeperServerPrincipal("zookeeper/hadoop.hadoop.com");
        LoginUtilKafka.setJaasFile(USER_PRINCIPAL, userKeyTableFile);
    }

    public static void run() {


        /*
         * 进行安全认证
         */
        if (isSecurityModel()) {
            try {
                LOG.info("Securitymode start.");
                // !!注意，安全认证时，需要用户手动修改为自己申请的机机账号
                securityPrepare();
            } catch (IOException e) {
                LOG.error("Security prepare failure.");
                LOG.error("The IOException occured.", e);
                return;
            }
            LOG.info("Security prepare success.");
        }


    }


}
